package top.wuchu.rpc.transport.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.wuchu.rpc.entity.RpcRequest;
import top.wuchu.rpc.entity.RpcResponse;
import top.wuchu.rpc.enumeration.RpcError;
import top.wuchu.rpc.exception.RpcException;
import top.wuchu.rpc.factory.SingletonFactory;
import top.wuchu.rpc.loadbalancer.LoadBalancer;
import top.wuchu.rpc.loadbalancer.RandomLoadBalancer;
import top.wuchu.rpc.registry.NacosServiceDiscovery;
import top.wuchu.rpc.registry.ServiceDiscovery;
import top.wuchu.rpc.serializer.CommonSerializer;
import top.wuchu.rpc.transport.RpcClient;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * NIO方式消费侧客户端类
 *
 * @author Wuchu
 */
public class NettyClient implements RpcClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    // EventLoop 定义了Netty的核心抽象，用来处理连接的生命周期中所发生的事件，在内部，将会为每个Channel分配一个EventLoop。
    // EventLoopGroup 是一个 EventLoop 池，包含很多的 EventLoop
    // Netty 为每个 Channel 分配了一个 EventLoop，用于处理用户连接请求、对用户请求的处理等所有事件。
    // EventLoop 本身只是一个线程驱动，在其生命周期内只会绑定一个线程，让该线程处理一个 Channel 的所有 IO 事件
    // 一个 Channel 一旦与一个 EventLoop 相绑定，那么在 Channel 的整个生命周期内是不能改变的。一个 EventLoop 可以与多个 Channel 绑定。
    // 即 Channel 与 EventLoop 的关系是 n:1，而 EventLoop 与线程的关系是 1:1

    private final ServiceDiscovery serviceDiscovery;
    private final CommonSerializer serializer;

    private final UnprocessedRequests unprocessedRequests;

    public NettyClient() {
        this(DEFAULT_SERIALIZER, new RandomLoadBalancer());
    }

    public NettyClient(LoadBalancer loadBalancer) {
        this(DEFAULT_SERIALIZER, loadBalancer);
    }

    public NettyClient(Integer serializer) {
        this(serializer, new RandomLoadBalancer());
    }

    public NettyClient(Integer serializer, LoadBalancer loadBalancer) {
        this.serviceDiscovery = new NacosServiceDiscovery(loadBalancer);
        this.serializer = CommonSerializer.getByCode(serializer);
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    /**
     * 向服务端发送请求
     * @param rpcRequest
     * @return
     */
    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            logger.error("未设置序列化器");
            throw new RpcException(RpcError.SERIALIZER_NOT_FOUND);
        }

        // 通过这个CompletableFuture异步回调的形式去处理客户端请求，最后通过resultFuture.get()拿到响应结果
        CompletableFuture<RpcResponse> resultFuture = new CompletableFuture<>();

        /**
         * 在上述实例化方法中，我们是可以指定Executor参数的，当我们不指定的试话，
         * 我们所开的并行线程使用的是默认系统及公共线程池ForkJoinPool，而且这些线程都是守护线程。
         */

        try {
            // 根据请求名称获取服务的IP+Port
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());

            // 根据IP+port 创建一个channel
            Channel channel = ChannelProvider.get(inetSocketAddress, serializer);

            // 每次发送一次请求，就存储在unprocessedRequests中
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);

            // 客户端将请求消息发送给服务端
            channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future1 -> {
                if (future1.isSuccess()) {
                    logger.info(String.format("客户端发送消息: %s", rpcRequest.toString()));
                } else {
                    future1.channel().close();
                    resultFuture.completeExceptionally(future1.cause());
                    logger.error("发送消息时有错误发生: ", future1.cause());
                }
            });
            System.out.println(111);


        } catch (InterruptedException e) {
            unprocessedRequests.remove(rpcRequest.getRequestId());
            logger.error(e.getMessage(), e);
            Thread.currentThread().interrupt();
        }
        return resultFuture;
    }

}
